package cn.jly.flink.wordcount;

import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

/**
 * @author lanyangji
 */
public class WordCountApp {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 为了测试效果，设置并行度为1
        env.setParallelism(1);

        ParameterTool parameterTool = ParameterTool.fromArgs(args);
        String host = parameterTool.get("host");
        int port = parameterTool.getInt("port");

        if (StringUtils.isEmpty(host) || port <= 0) {
            return;
        }

        env.socketTextStream(host, port)
                .flatMap(new LineToWordAndCount())
                .keyBy(0)
                //.reduce(new WordAnd1ToWordAndCount())
                .sum(1)
                .print(String.format("word count from socket %s:%d", host, port));

        env.execute("WordCountApp");
    }
}

/**
 * (word,1) -> (word, count)
 */
class WordAnd1ToWordAndCount implements ReduceFunction<Tuple2<String, Integer>> {

    @Override
    public Tuple2<String, Integer> reduce(Tuple2<String, Integer> value1, Tuple2<String, Integer> value2) throws Exception {
        return Tuple2.of(value1.f0, value1.f1 + value2.f1);
    }
}